/*
 * pgbench.c
 *
 * A simple benchmark program for openGauss
 * Originally written by Tatsuo Ishii and enhanced by many contributors.
 *
 * contrib/pgbench/pgbench.c
 * Copyright (c) 2000-2012, PostgreSQL Global Development Group
 * ALL RIGHTS RESERVED;
 *
 * Permission to use, copy, modify, and distribute this software and its
 * documentation for any purpose, without fee, and without a written agreement
 * is hereby granted, provided that the above copyright notice and this
 * paragraph and the following two paragraphs appear in all copies.
 *
 * IN NO EVENT SHALL THE AUTHOR OR DISTRIBUTORS BE LIABLE TO ANY PARTY FOR
 * DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES, INCLUDING
 * LOST PROFITS, ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS
 * DOCUMENTATION, EVEN IF THE AUTHOR OR DISTRIBUTORS HAVE BEEN ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 *
 * THE AUTHOR AND DISTRIBUTORS SPECIFICALLY DISCLAIMS ANY WARRANTIES,
 * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY
 * AND FITNESS FOR A PARTICULAR PURPOSE.  THE SOFTWARE PROVIDED HEREUNDER IS
 * ON AN "AS IS" BASIS, AND THE AUTHOR AND DISTRIBUTORS HAS NO OBLIGATIONS TO
 * PROVIDE MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS.
 *
 */

#ifdef WIN32
#define FD_SETSIZE 1024 /* set before winsock2.h is included */
#endif                  /* ! WIN32 */

#include "postgres_fe.h"

#include "getopt_long.h"
#include "libpq/libpq-fe.h"
#include "libpq/pqsignal.h"
#include "portability/instr_time.h"
#include "utils/elog.h"
#include <ctype.h>
#include <math.h>

#ifndef WIN32
#include <sys/time.h>
#include <unistd.h>
#endif /* ! WIN32 */

#ifdef HAVE_SYS_SELECT_H
#include <sys/select.h>
#endif

#ifdef HAVE_SYS_RESOURCE_H
#include <sys/resource.h> /* for getrlimit */
#endif

#ifdef HAVE_POLL_H
#include <poll.h>
#endif
#ifdef HAVE_SYS_POLL_H
#include <sys/poll.h>
#endif

#ifndef INT64_MAX
#define INT64_MAX INT64CONST(0x7FFFFFFFFFFFFFFF)
#endif

/*
 * Multi-platform pthread implementations
 */

#ifdef WIN32
/* Use native win32 threads on Windows */
typedef struct win32_pthread* pthread_t;
typedef int pthread_attr_t;

static int pthread_create(pthread_t* thread, pthread_attr_t* attr, void* (*start_routine)(void*), void* arg);
static int pthread_join(pthread_t th, void** thread_return);
#elif defined(ENABLE_THREAD_SAFETY)
/* Use platform-dependent pthread capability */
#include <pthread.h>
#else
/* Use emulation with fork. Rename pthread identifiers to avoid conflicts */
#define PTHREAD_FORK_EMULATION
#include <sys/wait.h>

#define pthread_t pg_pthread_t
#define pthread_attr_t pg_pthread_attr_t
#define pthread_create pg_pthread_create
#define pthread_join pg_pthread_join

typedef struct fork_pthread* pthread_t;
typedef int pthread_attr_t;

static int pthread_create(pthread_t* thread, pthread_attr_t* attr, void* (*start_routine)(void*), void* arg);
static int pthread_join(pthread_t th, void** thread_return);
#endif

extern char* optarg;
extern int optind;

/********************************************************************
 * some configurable parameters */

/* max number of clients allowed */
#ifdef FD_SETSIZE
#define MAXCLIENTS (FD_SETSIZE - 10)
#else
#define MAXCLIENTS 1024
#endif

#define DEFAULT_NXACTS 10 /* default nxacts */

int nxacts = 0;   /* number of transactions per client */
int duration = 0; /* duration in seconds */

/*
 * scaling factor. for example, scale = 10 will make 1000000 tuples in
 * pgbench_accounts table.
 */
int scale = 1;

/*
 * fillfactor. for example, fillfactor = 90 will use only 90 percent
 * space during inserts and leave 10 percent free.
 */
int fillfactor = 100;

/*
 * use unlogged tables?
 */
int unlogged_tables = 0;

/*
 * When threads are throttled to a given rate limit, this is the target delay
 * to reach that rate in usec.  0 is the default and means no throttling.
 */
int64 throttle_delay = 0;

/*
 * tablespace selection
 */
char* tablespace = NULL;
char* index_tablespace = NULL;

/*
 * end of configurable parameters
 *********************************************************************/

#define nbranches                                   \
    1 /* Makes little sense to change this.  Change \
       * -s instead */
#define ntellers 10
#define naccounts 100000

#ifdef PGXC
bool use_branch = false; /* use branch id in DDL and DML */
#endif
bool use_log;              /* log transaction latencies to a file */
int progress = 0;          /* thread progress report every this seconds */
int progress_nclients = 0; /* number of clients for progress report */
int progress_nthreads = 0; /* number of threads for progress report */
bool is_connect;           /* establish connection for each transaction */
bool is_mot = false;       /* use memory tables */
bool is_latencies;         /* report per-command latencies */
int main_pid;              /* main process id used in log filename */

char* pghost = "";
char* pgport = "";
char* pgoptions = NULL;
char* pgtty = NULL;
char* login = NULL;
char* secrete = NULL;
char* dbName;
char* orient = NULL;
char* with_options = NULL;

volatile bool timer_exceeded = false; /* flag from signal handler */
volatile int seq_curr_index = -1;

/* variable definitions */
typedef struct {
    char* name;  /* variable name */
    char* value; /* its value */
} Variable;

#define MAX_FILES 128          /* max number of SQL script files allowed */
#define SHELL_COMMAND_SIZE 256 /* maximum size allowed for shell command */

/*
 * structures used in custom query mode
 */

typedef struct {
    PGconn* con;         /* connection handle to DB */
    int id;              /* client No. */
    int state;           /* state No. */
    int cnt;             /* xacts count */
    int ecnt;            /* error count */
    int listen;          /* 0 indicates that an async query has been
                          * sent */
    int sleeping;        /* 1 indicates that the client is napping */
    bool throttling;     /* whether nap is for throttling */
    uint64 until;        /* napping until (usec) */
    Variable* variables; /* array of variable definitions */
    int nvariables;
    instr_time txn_begin;  /* used for measuring transaction latencies */
    instr_time stmt_begin; /* used for measuring statement latencies */
    int64 txn_latencies;   /* cumulated latencies */
    int64 txn_sqlats;      /* cumulated square latencies */
    bool is_throttled;     /* whether transaction throttling is done */
    int use_file;          /* index in sql_files for this client */
    bool prepared[MAX_FILES];
} CState;

/*
 * Thread state and result
 */
typedef struct {
    int tid;                        /* thread id */
    pthread_t thread;               /* thread handle */
    CState* state;                  /* array of CState */
    int nstate;                     /* length of state[] */
    instr_time start_time;          /* thread start time */
    instr_time* exec_elapsed;       /* time spent executing cmds (per Command) */
    int* exec_count;                /* number of cmd executions (per Command) */
    unsigned short random_state[3]; /* separate randomness for each thread */
    int64 throttle_trigger;         /* previous/next throttling (us) */
    int64 throttle_lag;             /* total transaction lag behind throttling */
    int64 throttle_lag_max;         /* max transaction lag */
} TState;

#define INVALID_THREAD ((pthread_t)0)

typedef struct {
    instr_time conn_time;
    int xacts;
    int64 latencies;
    int64 sqlats;
    int64 throttle_lag;
    int64 throttle_lag_max;
} TResult;

/*
 * queries read from files
 */
#define SQL_COMMAND 1
#define META_COMMAND 2
#define MAX_ARGS 32

typedef enum QueryMode {
    QUERY_SIMPLE,   /* simple query */
    QUERY_EXTENDED, /* extended query */
    QUERY_PREPARED, /* extended query with prepared statements */
    NUM_QUERYMODE
} QueryMode;

static QueryMode querymode = QUERY_SIMPLE;
static const char* QUERYMODE[] = {"simple", "extended", "prepared"};

typedef struct {
    char* line;           /* full text of command line */
    int command_num;      /* unique index of this Command struct */
    int type;             /* command type (SQL_COMMAND or META_COMMAND) */
    int argc;             /* number of command words */
    char* argv[MAX_ARGS]; /* command word list */
} Command;

static Command** sql_files[MAX_FILES]; /* SQL script files */
static int num_files;                  /* number of script files */
static int num_commands = 0;           /* total number of Command structs */
static int debug = 0;                  /* debug flag */

/* default scenario */
static char* tpc_b = {"\\set nbranches " CppAsString2(
    nbranches) " * :scale\n"
               "\\set ntellers " CppAsString2(
                   ntellers) " * :scale\n"
                             "\\set naccounts " CppAsString2(
                                 naccounts) " * :scale\n"
                                            "\\setrandom aid 1 :naccounts\n"
                                            "\\setrandom bid 1 :nbranches\n"
                                            "\\setrandom ttid 1 :ntellers\n"
                                            "\\setrandom delta -5000 5000\n"
                                            "START TRANSACTION;\n"
                                            "UPDATE pgbench_accounts SET abalance = abalance + :delta WHERE aid = "
                                            ":aid;\n"
                                            "SELECT abalance FROM pgbench_accounts WHERE aid = :aid;\n"
                                            "UPDATE pgbench_tellers SET tbalance = tbalance + :delta WHERE ttid = "
                                            ":ttid;\n"
                                            "UPDATE pgbench_branches SET bbalance = bbalance + :delta WHERE bid = "
                                            ":bid;\n"
                                            "INSERT INTO pgbench_history (ttid, bid, aid, delta, mtime) VALUES (:ttid, "
                                            ":bid, :aid, :delta, CURRENT_TIMESTAMP);\n"
                                            "END;\n"};

#ifdef PGXC
static char* tpc_b_bid = {"\\set nbranches " CppAsString2(
    nbranches) " * :scale\n"
               "\\set ntellers " CppAsString2(
                   ntellers) " * :scale\n"
                             "\\set naccounts " CppAsString2(
                                 naccounts) " * :scale\n"
                                            "\\setrandom aid 1 :naccounts\n"
                                            "\\setrandom bid 1 :nbranches\n"
                                            "\\setrandom ttid 1 :ntellers\n"
                                            "\\setrandom delta -5000 5000\n"
                                            "START TRANSACTION;\n"
                                            "UPDATE pgbench_accounts SET abalance = abalance + :delta WHERE aid = :aid "
                                            "AND bid = :bid;\n"
                                            "SELECT abalance FROM pgbench_accounts WHERE aid = :aid AND bid = :bid\n"
                                            "UPDATE pgbench_tellers SET tbalance = tbalance + :delta WHERE ttid = "
                                            ":ttid AND bid = :bid;\n"
                                            "UPDATE pgbench_branches SET bbalance = bbalance + :delta WHERE bid = "
                                            ":bid;\n"
                                            "INSERT INTO pgbench_history (ttid, bid, aid, delta, mtime) VALUES (:ttid, "
                                            ":bid, :aid, :delta, CURRENT_TIMESTAMP);\n"
                                            "END;\n"};
#endif

/* -N case */
static char* simple_update = {"\\set nbranches " CppAsString2(
    nbranches) " * :scale\n"
               "\\set ntellers " CppAsString2(
                   ntellers) " * :scale\n"
                             "\\set naccounts " CppAsString2(
                                 naccounts) " * :scale\n"
                                            "\\setrandom aid 1 :naccounts\n"
                                            "\\setrandom bid 1 :nbranches\n"
                                            "\\setrandom ttid 1 :ntellers\n"
                                            "\\setrandom delta -5000 5000\n"
                                            "START TRANSACTION;\n"
                                            "UPDATE pgbench_accounts SET abalance = abalance + :delta WHERE aid = "
                                            ":aid;\n"
                                            "SELECT abalance FROM pgbench_accounts WHERE aid = :aid;\n"
                                            "INSERT INTO pgbench_history (ttid, bid, aid, delta, mtime) VALUES (:ttid, "
                                            ":bid, :aid, :delta, CURRENT_TIMESTAMP);\n"
                                            "END;\n"};

#ifdef PGXC
static char* simple_update_bid = {"\\set nbranches " CppAsString2(
    nbranches) " * :scale\n"
               "\\set ntellers " CppAsString2(
                   ntellers) " * :scale\n"
                             "\\set naccounts " CppAsString2(
                                 naccounts) " * :scale\n"
                                            "\\setrandom aid 1 :naccounts\n"
                                            "\\setrandom bid 1 :nbranches\n"
                                            "\\setrandom ttid 1 :ntellers\n"
                                            "\\setrandom delta -5000 5000\n"
                                            "START TRANSACTION;\n"
                                            "UPDATE pgbench_accounts SET abalance = abalance + :delta WHERE aid = :aid "
                                            "AND bid = :bid;\n"
                                            "SELECT abalance FROM pgbench_accounts WHERE aid = :aid AND bid = :bid;\n"
                                            "INSERT INTO pgbench_history (ttid, bid, aid, delta, mtime) VALUES (:ttid, "
                                            ":bid, :aid, :delta, CURRENT_TIMESTAMP);\n"
                                            "END;\n"};
#endif

/* -S case */
static char* select_only = {
    "\\set naccounts " CppAsString2(naccounts) " * :scale\n"
                                               "\\setrandom aid 1 :naccounts\n"
                                               "SELECT abalance FROM pgbench_accounts WHERE aid = :aid;\n"};

/* Function prototypes */
static void setalarm(int seconds);
static void* threadRun(void* arg);

/*
 * routines to check mem allocations and fail noisily.
 */
static void* xmalloc(size_t size)
{
    void* result = NULL;

    /* Avoid unportable behavior of malloc(0) */
    if (size == 0) {
        size = 1;
    }
    result = malloc(size);
    if (!result) {
        fprintf(stderr, "out of memory\n");
        exit(1);
    }
    return result;
}

static void* xrealloc(void* ptr, size_t size)
{
    void* result = NULL;

    /* Avoid unportable behavior of realloc(NULL, 0) */
    if (ptr == NULL && size == 0) {
        size = 1;
    }
    result = realloc(ptr, size);
    if (!result) {
        fprintf(stderr, "out of memory\n");
        exit(1);
    }
    return result;
}

static char* xstrdup(const char* s)
{
    char* result = NULL;

    result = strdup(s);
    if (!result) {
        fprintf(stderr, "out of memory\n");
        exit(1);
    }
    return result;
}

static void usage(const char* progname)
{
    printf("%s is a benchmarking tool for openGauss.\n\n"
           "Usage:\n"
           "  %s [OPTION]... [DBNAME]\n"
           "\nInitialization options:\n"
           "  -i           invokes initialization mode\n"
           "  -m           use memory tables (mot)\n"
           "  -F NUM       fill factor\n"
#ifdef PGXC
           "  -k           distribute by primary key branch id - bid\n"
#endif
           "  -s NUM       scaling factor\n"
           "  --index-tablespace=TABLESPACE\n"
           "               create indexes in the specified tablespace\n"
           "  --tablespace=TABLESPACE\n"
           "               create tables in the specified tablespace\n"
           "  --unlogged-tables\n"
           "               create tables as unlogged tables\n"
           "\nBenchmarking options:\n"
           "  -c NUM       number of concurrent database clients (default: 1)\n"
           "  -C           establish new connection for each transaction\n"
           "  -D VARNAME=VALUE\n"
           "               define variable for use by custom script\n"
           "  -f FILENAME  read transaction script from FILENAME\n"
#ifdef PGXC
           "  -k           query with default key and additional key branch id (bid)\n"
#endif
           "  -j NUM       number of threads (default: 1)\n"
           "  -l           write transaction times to log file\n"
           "  -M simple|extended|prepared\n"
           "               protocol for submitting queries to server (default: simple)\n"
           "  -n           do not run VACUUM before tests\n"
           "  -N           do not update tables \"pgbench_tellers\" and \"pgbench_branches\"\n"
           "  -O row|column|orc\n"
           "               Table orientation option\n"
           "  -P NUM       show thread progress report every NUM seconds\n"
           "  -r           report average latency per command\n"
           "  -R, --rate=NUM\n"
           "               target rate in transactions per second\n"
           "  -s NUM       report this scale factor in output\n"
           "  -S           perform SELECT-only transactions\n"
           "  -t NUM       number of transactions each client runs (default: 10)\n"
           "  -T NUM       duration of benchmark test in seconds\n"
           "  -v           vacuum all four standard tables before tests\n"
           "\nCommon options:\n"
           "  -d             print debugging output\n"
           "  -h HOSTNAME    database server host or socket directory\n"
           "  -p PORT        database server port number\n"
           "  -U USERNAME    connect as specified database user\n"
           "  -W PASSWORD    connect as specified database user through explicit password\n"
           "  -V, --version  output version information, then exit\n"
           "  -?, --help     show this help, then exit\n",
        progname,
        progname);
#if ((defined(ENABLE_MULTIPLE_NODES)) || (defined(ENABLE_PRIVATEGAUSS)))
    printf("\nReport bugs to GaussDB support.\n");
#else
    printf("\nReport bugs to community@opengauss.org> or join opengauss community <https://opengauss.org>.\n");
#endif
}

/* random number generator: uniform distribution from min to max inclusive */
static int getrand(TState* thread, int min, int max)
{
    /*
     * Odd coding is so that min and max have approximately the same chance of
     * being selected as do numbers between them.
     *
     * pg_erand48() is thread-safe and concurrent, which is why we use it
     * rather than random(), which in glibc is non-reentrant, and therefore
     * protected by a mutex, and therefore a bottleneck on machines with many
     * CPUs.
     */
    return min + (int)((max - min + 1) * pg_erand48(thread->random_state));
}

/* call PQexec() and exit() on failure */
static void executeStatement(PGconn* con, const char* sql, bool err_tolerant = false)
{
    PGresult* res = NULL;

    res = PQexec(con, sql);
    if (PQresultStatus(res) != PGRES_COMMAND_OK) {
        fprintf(stderr, "%s", PQerrorMessage(con));
        if (!err_tolerant)
            exit(1);
    }
    PQclear(res);
}

/* set up a connection to the backend */
static PGconn* doConnect(void)
{
    PGconn* conn = NULL;
    static char* password = NULL;
    bool new_pass = false;

    if (secrete != NULL) {
        password = secrete;
    }
    /*
     * Start the connection.  Loop until we have a password if requested by
     * backend.
     */
    do {
        new_pass = false;

        conn = PQsetdbLogin(pghost, pgport, pgoptions, pgtty, dbName, login, password);
        if (!conn) {
            fprintf(stderr, "Connection to database \"%s\" failed\n", dbName);
            return NULL;
        }

        if (PQstatus(conn) == CONNECTION_BAD && PQconnectionNeedsPassword(conn) && password == NULL) {
            PQfinish(conn);
            password = simple_prompt("Password: ", 100, false);
            new_pass = true;
        }
    } while (new_pass);

    /* check to see that the backend connection was successfully made */
    if (PQstatus(conn) == CONNECTION_BAD) {
        fprintf(stderr, "Connection to database \"%s\" failed:\n%s", dbName, PQerrorMessage(conn));
        PQfinish(conn);
        return NULL;
    }

    return conn;
}

/* throw away response from backend */
static void discard_response(CState* state)
{
    PGresult* res = NULL;

    do {
        res = PQgetResult(state->con);
        if (res)
            PQclear(res);
    } while (res);
}

static int compareVariables(const void* v1, const void* v2)
{
    return strcmp(((const Variable*)v1)->name, ((const Variable*)v2)->name);
}

static char* getVariable(CState* st, char* name)
{
    Variable key;
    Variable* var = NULL;

    /* On some versions of Solaris, bsearch of zero items dumps core */
    if (st->nvariables <= 0) {
        return NULL;
    }

    key.name = name;
    var = (Variable*)bsearch((void*)&key, (void*)st->variables, st->nvariables, sizeof(Variable), compareVariables);
    if (var != NULL) {
        return var->value;
    } else {
        return NULL;
    }
}

/* check whether the name consists of alphabets, numerals and underscores. */
static bool isLegalVariableName(const char* name)
{
    int i;

    for (i = 0; name[i] != '\0'; i++) {
        if (!isalnum((unsigned char)name[i]) && name[i] != '_') {
            return false;
        }
    }

    return true;
}

static int putVariable(CState* st, const char* context, char* name, char* value)
{
    Variable key;
    Variable* var = NULL;

    key.name = name;
    /* On some versions of Solaris, bsearch of zero items dumps core */
    if (st->nvariables > 0) {
        var = (Variable*)bsearch((void*)&key, (void*)st->variables, st->nvariables, sizeof(Variable), compareVariables);
    } else {
        var = NULL;
    }

    if (var == NULL) {
        Variable* newvars = NULL;

        /*
         * Check for the name only when declaring a new variable to avoid
         * overhead.
         */
        if (!isLegalVariableName(name)) {
            fprintf(stderr, "%s: invalid variable name '%s'\n", context, name);
            return false;
        }

        if (st->variables) {
            newvars = (Variable*)xrealloc(st->variables, (st->nvariables + 1) * sizeof(Variable));
        } else {
            newvars = (Variable*)xmalloc(sizeof(Variable));
        }

        st->variables = newvars;

        var = &newvars[st->nvariables];

        var->name = xstrdup(name);
        var->value = xstrdup(value);

        st->nvariables++;

        qsort((void*)st->variables, st->nvariables, sizeof(Variable), compareVariables);
    } else {
        char* val = NULL;

        /* dup then free, in case value is pointing at this variable */
        val = xstrdup(value);

        free(var->value);
        var->value = val;
    }

    return true;
}

static char* parseVariable(const char* sql, int* eaten)
{
    int i = 0;
    char* name = NULL;

    do {
        i++;
    } while (isalnum((unsigned char)sql[i]) || sql[i] == '_');
    if (i == 1) {
        return NULL;
    }

    name = (char*)xmalloc(i);
    errno_t rc = memcpy_s(name, i, &sql[1], i - 1);
    securec_check_c(rc, "", "");
    name[i - 1] = '\0';

    *eaten = i;
    return name;
}

static char* replaceVariable(char** sql, char* param, int len, char* value)
{
    int valueln = strlen(value);
    errno_t rc;

    if (valueln > len) {
        size_t offset = param - *sql;

        *sql = (char*)xrealloc(*sql, strlen(*sql) - len + valueln + 1);
        param = *sql + offset;
    }

    if (valueln != len)
        memmove(param + valueln, param + len, strlen(param + len) + 1);
    rc = memcpy_s(param, valueln, value, valueln);
    securec_check_c(rc, "", "");

    return param + valueln;
}

static char* assignVariables(CState* st, char* sql)
{
    char *p = NULL;
    char *name = NULL;
    char *val = NULL;

    p = sql;
    while ((p = strchr(p, ':')) != NULL) {
        int eaten;

        name = parseVariable(p, &eaten);
        if (name == NULL) {
            while (*p == ':') {
                p++;
            }
            continue;
        }

        val = getVariable(st, name);
        free(name);
        if (val == NULL) {
            p++;
            continue;
        }

        p = replaceVariable(&sql, p, eaten, val);
    }

    return sql;
}

static void getQueryParams(CState* st, const Command* command, const char** params)
{
    int i;

    for (i = 0; i < command->argc - 1; i++)
        params[i] = getVariable(st, command->argv[i + 1]);
}

/*
 * Run a shell command. The result is assigned to the variable if not NULL.
 * Return true if succeeded, or false on error.
 */
static bool runShellCommand(CState* st, char* variable, char** argv, int argc)
{
    char command[SHELL_COMMAND_SIZE];
    int i, len = 0;
    FILE* fp = NULL;
    char res[64];
    char* endptr = NULL;
    int retval;

    /*----------
     * Join arguments with whitespace separators. Arguments starting with
     * exactly one colon are treated as variables:
     *	name - append a string "name"
     *	:var - append a variable named 'var'
     *	::name - append a string ":name"
     *----------
     */
    for (i = 0; i < argc; i++) {
        char* arg = NULL;
        int arglen;

        if (argv[i][0] != ':') {
            arg = argv[i]; /* a string literal */
        } else if (argv[i][1] == ':') {
            arg = argv[i] + 1; /* a string literal starting with colons */
        } else if ((arg = getVariable(st, argv[i] + 1)) == NULL) {
            fprintf(stderr, "%s: undefined variable %s\n", argv[0], argv[i]);
            return false;
        }

        arglen = strlen(arg);
        if ((len + arglen + ((i > 0) ? 1 : 0)) >= (SHELL_COMMAND_SIZE - 1)) {
            fprintf(stderr, "%s: too long shell command\n", argv[0]);
            return false;
        }

        if (i > 0)
            command[len++] = ' ';
        errno_t rc = memcpy_s(command + len, SHELL_COMMAND_SIZE - len, arg, arglen);
        securec_check_c(rc, "", "");
        len += arglen;
    }

    command[len] = '\0';

    /* Fast path for non-assignment case */
    if (variable == NULL) {
        if (system(command)) {
            if (!timer_exceeded)
                fprintf(stderr, "%s: cannot launch shell command\n", argv[0]);
            return false;
        }
        return true;
    }

    /* Execute the command with pipe and read the standard output. */
    if ((fp = popen(command, "r")) == NULL) {
        fprintf(stderr, "%s: cannot launch shell command\n", argv[0]);
        return false;
    }
    if (fgets(res, sizeof(res), fp) == NULL) {
        if (!timer_exceeded)
            fprintf(stderr, "%s: cannot read the result\n", argv[0]);
        return false;
    }
    if (pclose(fp) < 0) {
        fprintf(stderr, "%s: cannot close shell command\n", argv[0]);
        return false;
    }

    /* Check whether the result is an integer and assign it to the variable */
    retval = (int)strtol(res, &endptr, 10);
    while (*endptr != '\0' && isspace((unsigned char)*endptr)) {
        endptr++;
    }
    if (*res == '\0' || *endptr != '\0') {
        fprintf(stderr, "%s: must return an integer ('%s' returned)\n", argv[0], res);
        return false;
    }
    snprintf(res, sizeof(res), "%d", retval);
    if (!putVariable(st, "setshell", variable, res))
        return false;

#ifdef DEBUG
    printf("shell parameter name: %s, value: %s\n", argv[1], res);
#endif
    return true;
}

#define MAX_PREPARE_NAME 32
static void preparedStatementName(char* buffer, int file, int state)
{
    sprintf(buffer, "P%d_%d", file, state);
}

static bool clientDone(CState* st, bool ok)
{
    (void)ok; /* unused */

    if (st->con != NULL) {
        PQfinish(st->con);
        st->con = NULL;
    }
    return false; /* always false */
}

/* return false if client should be disconnected */
static bool doCustom(TState* thread, CState* st, instr_time* conn_time, FILE* logfile)
{
    PGresult* res = NULL;
    Command** commands;
    bool trans_needs_throttle = false;

top:
    commands = sql_files[st->use_file];

    /*
     * Handle throttling once per transaction by sleeping.  It is simpler
     * to do this here rather than at the end, because so much complicated
     * logic happens below when statements finish.
     */
    if (throttle_delay && !st->is_throttled) {
        /*
         * Use inverse transform sampling to randomly generate a delay, such
         * that the series of delays will approximate a Poisson distribution
         * centered on the throttle_delay time.
         *
         * 10000 implies a 9.2 (-log(1/10000)) to 0.0 (log 1) delay multiplier,
         * and results in a 0.055 % target underestimation bias:
         *
         * SELECT 1.0/AVG(-LN(i/10000.0)) FROM generate_series(1,10000) AS i;
         * = 1.000552717032611116335474
         *
         * If transactions are too slow or a given wait is shorter than
         * a transaction, the next transaction will start right away.
         */
        int64 wait = (int64)(throttle_delay * 1.00055271703 * -log(getrand(thread, 1, 10000) / 10000.0));

        thread->throttle_trigger += wait;

        st->until = thread->throttle_trigger;
        st->sleeping = 1;
        st->throttling = true;
        st->is_throttled = true;
        if (debug)
            fprintf(stderr, "client %d throttling " INT64_FORMAT " us\n", st->id, wait);
    }

    if (st->sleeping) { /* are we sleeping? */
        instr_time now;
        int64 now_us;

        INSTR_TIME_SET_CURRENT(now);
        now_us = INSTR_TIME_GET_MICROSEC(now);
        if (st->until <= (uint64)now_us) {
            st->sleeping = 0; /* Done sleeping, go ahead with next command */
            if (st->throttling) {
                /* Measure lag of throttled transaction relative to target */
                int64 lag = now_us - st->until;
                thread->throttle_lag += lag;
                if (lag > thread->throttle_lag_max)
                    thread->throttle_lag_max = lag;
                st->throttling = false;
            }
        } else
            return true; /* Still sleeping, nothing to do here */
    }

    if (st->listen) { /* are we receiver? */
        if (commands[st->state]->type == SQL_COMMAND) {
            if (debug)
                fprintf(stderr, "client %d receiving\n", st->id);
            if (!PQconsumeInput(st->con)) { /* there's something wrong */
                fprintf(stderr,
                    "Client %d aborted in state %d. Probably the backend died while processing.\n",
                    st->id,
                    st->state);
                return clientDone(st, false);
            }
            if (PQisBusy(st->con))
                return true; /* don't have the whole result yet */
        }

        /*
         * command finished: accumulate per-command execution times in
         * thread-local data structure, if per-command latencies are requested
         */
        if (is_latencies) {
            instr_time now;
            int cnum = commands[st->state]->command_num;

            INSTR_TIME_SET_CURRENT(now);
            INSTR_TIME_ACCUM_DIFF(thread->exec_elapsed[cnum], now, st->stmt_begin);
            thread->exec_count[cnum]++;
        }

        /* transaction finished: record latency under progress or throttling */
        if ((progress || throttle_delay) && commands[st->state + 1] == NULL) {
            instr_time diff;
            int64 latency;

            INSTR_TIME_SET_CURRENT(diff);
            INSTR_TIME_SUBTRACT(diff, st->txn_begin);
            latency = INSTR_TIME_GET_MICROSEC(diff);
            st->txn_latencies += latency;
            /*
             * XXX In a long benchmark run of high-latency transactions, this
             * int64 addition eventually overflows.  For example, 100 threads
             * running 10s transactions will overflow it in 2.56 hours.  With
             * a more-typical OLTP workload of .1s transactions, overflow
             * would take 256 hours.
             */
            st->txn_sqlats += latency * latency;
        }

        /*
         * if transaction finished, record the time it took in the log
         */
        if (logfile && commands[st->state + 1] == NULL) {
            instr_time now;
            instr_time diff;
            double usec;

            INSTR_TIME_SET_CURRENT(now);
            diff = now;
            INSTR_TIME_SUBTRACT(diff, st->txn_begin);
            usec = (double)INSTR_TIME_GET_MICROSEC(diff);

#ifndef WIN32
            /* This is more than we really ought to know about instr_time */
            fprintf(logfile,
                "%d %d %.0f %d %ld %ld\n",
                st->id,
                st->cnt,
                usec,
                st->use_file,
                (long)now.tv_sec,
                (long)now.tv_usec);
#else
            /* On Windows, instr_time doesn't provide a timestamp anyway */
            fprintf(logfile, "%d %d %.0f %d 0 0\n", st->id, st->cnt, usec, st->use_file);
#endif
        }

        /* identify transaction errors */
        bool error_found = false;

        if (commands[st->state]->type == SQL_COMMAND) {
            /*
             * Read and discard the query result; note this is not included in
             * the statement latency numbers.
             */
            res = PQgetResult(st->con);
            switch (PQresultStatus(res)) {
                case PGRES_COMMAND_OK:
                case PGRES_TUPLES_OK:
                    break; /* OK */
                default:
                    if (!is_mot) {
                        fprintf(
                            stderr, "Client %d aborted in state %d: %s", st->id, st->state, PQerrorMessage(st->con));
                    } else {
                        error_found = true;
                    }
                    break;
            }
            PQclear(res);
            discard_response(st);
        }

        if (commands[st->state + 1] == NULL) {
            if (is_connect) {
                PQfinish(st->con);
                st->con = NULL;
            }

            if (!error_found) {
                ++st->cnt;
            }

            if ((st->cnt >= nxacts && duration <= 0) || timer_exceeded)
                return clientDone(st, true); /* exit success */
        }

        /* increment state counter */
        st->state++;
        if (commands[st->state] == NULL) {
            st->state = 0;
            st->use_file = getrand(thread, 0, num_files - 1);
            commands = sql_files[st->use_file];
            st->is_throttled = false;
            /*
             * No transaction is underway anymore, which means there is nothing
             * to listen to right now.  When throttling rate limits are active,
             * a sleep will happen next, as the next transaction starts.  And
             * then in any case the next SQL command will set listen back to 1.
             */
            st->listen = 0;
            trans_needs_throttle = (throttle_delay > 0);
        }
    }

    if (st->con == NULL) {
        instr_time start, end;

        INSTR_TIME_SET_CURRENT(start);
        if ((st->con = doConnect()) == NULL) {
            fprintf(stderr, "Client %d aborted in establishing connection.\n", st->id);
            return clientDone(st, false);
        }
        INSTR_TIME_SET_CURRENT(end);
        INSTR_TIME_ACCUM_DIFF(*conn_time, end, start);
    }

    /*
     * This ensures that a throttling delay is inserted before proceeding
     * with sql commands, after the first transaction. The first transaction
     * throttling is performed when first entering doCustom.
     */
    if (trans_needs_throttle) {
        trans_needs_throttle = false;
        goto top;
    }

    /* Record transaction start time under logging, progress or throttling */
    if ((logfile || progress || throttle_delay) && st->state == 0)
        INSTR_TIME_SET_CURRENT(st->txn_begin);

    /* Record statement start time if per-command latencies are requested */
    if (is_latencies)
        INSTR_TIME_SET_CURRENT(st->stmt_begin);

    if (commands[st->state]->type == SQL_COMMAND) {
        const Command* command = commands[st->state];
        int r;

        if (querymode == QUERY_SIMPLE) {
            char* sql = NULL;

            sql = xstrdup(command->argv[0]);
            sql = assignVariables(st, sql);

            if (debug)
                fprintf(stderr, "client %d sending %s\n", st->id, sql);
            r = PQsendQuery(st->con, sql);
            free(sql);
        } else if (querymode == QUERY_EXTENDED) {
            const char* sql = command->argv[0];
            const char* params[MAX_ARGS];

            getQueryParams(st, command, params);

            if (debug)
                fprintf(stderr, "client %d sending %s\n", st->id, sql);
            r = PQsendQueryParams(st->con, sql, command->argc - 1, NULL, params, NULL, NULL, 0);
        } else if (querymode == QUERY_PREPARED) {
            char name[MAX_PREPARE_NAME];
            const char* params[MAX_ARGS];

            if (!st->prepared[st->use_file]) {
                int j;

                for (j = 0; commands[j] != NULL; j++) {
                    PGresult* res = NULL;
                    char name[MAX_PREPARE_NAME];

                    if (commands[j]->type != SQL_COMMAND) {
                        continue;
                    }
                    preparedStatementName(name, st->use_file, j);
                    res = PQprepare(st->con, name, commands[j]->argv[0], commands[j]->argc - 1, NULL);
                    if (PQresultStatus(res) != PGRES_COMMAND_OK)
                        fprintf(stderr, "%s", PQerrorMessage(st->con));
                    PQclear(res);
                }
                st->prepared[st->use_file] = true;
            }

            getQueryParams(st, command, params);
            preparedStatementName(name, st->use_file, st->state);

            if (debug)
                fprintf(stderr, "client %d sending %s\n", st->id, name);
            r = PQsendQueryPrepared(st->con, name, command->argc - 1, params, NULL, NULL, 0);
        } else /* unknown sql mode */
            r = 0;

        if (r == 0) {
            if (debug)
                fprintf(stderr, "client %d cannot send %s\n", st->id, command->argv[0]);
            st->ecnt++;
        } else
            st->listen = 1; /* flags that should be listened */
    } else if (commands[st->state]->type == META_COMMAND) {
        int argc = commands[st->state]->argc, i;
        char** argv = commands[st->state]->argv;

        if (debug) {
            fprintf(stderr, "client %d executing \\%s", st->id, argv[0]);
            for (i = 1; i < argc; i++)
                fprintf(stderr, " %s", argv[i]);
            fprintf(stderr, "\n");
        }

        if (pg_strcasecmp(argv[0], "setrandom") == 0) {
            char* var = NULL;
            int min, max;
            char res[64];

            if (*argv[2] == ':') {
                if ((var = getVariable(st, argv[2] + 1)) == NULL) {
                    fprintf(stderr, "%s: undefined variable %s\n", argv[0], argv[2]);
                    st->ecnt++;
                    return true;
                }
                min = atoi(var);
            } else
                min = atoi(argv[2]);

#ifdef NOT_USED
            if (min < 0) {
                fprintf(stderr, "%s: invalid minimum number %d\n", argv[0], min);
                st->ecnt++;
                return;
            }
#endif

            if (*argv[3] == ':') {
                if ((var = getVariable(st, argv[3] + 1)) == NULL) {
                    fprintf(stderr, "%s: undefined variable %s\n", argv[0], argv[3]);
                    st->ecnt++;
                    return true;
                }
                max = atoi(var);
            } else
                max = atoi(argv[3]);

            if (max < min) {
                fprintf(stderr, "%s: maximum is less than minimum\n", argv[0]);
                st->ecnt++;
                return true;
            }

            /*
             * getrand() neeeds to be able to subtract max from min and add
             * one the result without overflowing.	Since we know max > min,
             * we can detect overflow just by checking for a negative result.
             * But we must check both that the subtraction doesn't overflow,
             * and that adding one to the result doesn't overflow either.
             */
            if (max - min < 0 || (max - min) + 1 < 0) {
                fprintf(stderr, "%s: range too large\n", argv[0]);
                st->ecnt++;
                return true;
            }

#ifdef DEBUG
            printf("min: %d max: %d random: %d\n", min, max, getrand(thread, min, max));
#endif
            snprintf(res, sizeof(res), "%d", getrand(thread, min, max));

            if (!putVariable(st, argv[0], argv[1], res)) {
                st->ecnt++;
                return true;
            }

            st->listen = 1;
        } else if (pg_strcasecmp(argv[0], "setseq") == 0) {
            char* var = NULL;
            int min, max;
            char res[64];

            if (*argv[2] == ':') {
                if ((var = getVariable(st, argv[2] + 1)) == NULL) {
                    fprintf(stderr, "%s: undefined variable %s\n", argv[0], argv[2]);
                    st->ecnt++;
                    return true;
                }
                min = atoi(var);
            } else
                min = atoi(argv[2]);

#ifdef NOT_USED
            if (min < 0) {
                fprintf(stderr, "%s: invalid minimum number %d\n", argv[0], min);
                st->ecnt++;
                return;
            }
#endif

            if (*argv[3] == ':') {
                if ((var = getVariable(st, argv[3] + 1)) == NULL) {
                    fprintf(stderr, "%s: undefined variable %s\n", argv[0], argv[3]);
                    st->ecnt++;
                    return true;
                }
                max = atoi(var);
            } else
                max = atoi(argv[3]);

            if (max < min) {
                fprintf(stderr, "%s: maximum is less than minimum\n", argv[0]);
                st->ecnt++;
                return true;
            }

            /*
             * getrand() neeeds to be able to subtract max from min and add
             * one the result without overflowing.	Since we know max > min,
             * we can detect overflow just by checking for a negative result.
             * But we must check both that the subtraction doesn't overflow,
             * and that adding one to the result doesn't overflow either.
             */
            if (max - min < 0 || (max - min) + 1 < 0) {
                fprintf(stderr, "%s: range too large\n", argv[0]);
                st->ecnt++;
                return true;
            }

            seq_curr_index++;
            if (seq_curr_index > max)
                seq_curr_index = min;
            if (seq_curr_index < min)
                seq_curr_index = min;

            snprintf(res, sizeof(res), "%d", seq_curr_index);

            if (!putVariable(st, argv[0], argv[1], res)) {
                st->ecnt++;
                return true;
            }

            st->listen = 1;
        } else if (pg_strcasecmp(argv[0], "set") == 0) {
            char* var = NULL;
            int ope1, ope2;
            char res[64];

            if (*argv[2] == ':') {
                if ((var = getVariable(st, argv[2] + 1)) == NULL) {
                    fprintf(stderr, "%s: undefined variable %s\n", argv[0], argv[2]);
                    st->ecnt++;
                    return true;
                }
                ope1 = atoi(var);
            } else
                ope1 = atoi(argv[2]);

            if (argc < 5)
                snprintf(res, sizeof(res), "%d", ope1);
            else {
                if (*argv[4] == ':') {
                    if ((var = getVariable(st, argv[4] + 1)) == NULL) {
                        fprintf(stderr, "%s: undefined variable %s\n", argv[0], argv[4]);
                        st->ecnt++;
                        return true;
                    }
                    ope2 = atoi(var);
                } else
                    ope2 = atoi(argv[4]);

                if (strcmp(argv[3], "+") == 0)
                    snprintf(res, sizeof(res), "%d", ope1 + ope2);
                else if (strcmp(argv[3], "-") == 0)
                    snprintf(res, sizeof(res), "%d", ope1 - ope2);
                else if (strcmp(argv[3], "*") == 0)
                    snprintf(res, sizeof(res), "%d", ope1 * ope2);
                else if (strcmp(argv[3], "/") == 0) {
                    if (ope2 == 0) {
                        fprintf(stderr, "%s: division by zero\n", argv[0]);
                        st->ecnt++;
                        return true;
                    }
                    snprintf(res, sizeof(res), "%d", ope1 / ope2);
                } else {
                    fprintf(stderr, "%s: unsupported operator %s\n", argv[0], argv[3]);
                    st->ecnt++;
                    return true;
                }
            }

            if (!putVariable(st, argv[0], argv[1], res)) {
                st->ecnt++;
                return true;
            }

            st->listen = 1;
        } else if (pg_strcasecmp(argv[0], "sleep") == 0) {
            char* var = NULL;
            int usec;
            instr_time now;

            if (*argv[1] == ':') {
                if ((var = getVariable(st, argv[1] + 1)) == NULL) {
                    fprintf(stderr, "%s: undefined variable %s\n", argv[0], argv[1]);
                    st->ecnt++;
                    return true;
                }
                usec = atoi(var);
            } else
                usec = atoi(argv[1]);

            if (argc > 2) {
                if (pg_strcasecmp(argv[2], "ms") == 0)
                    usec *= 1000;
                else if (pg_strcasecmp(argv[2], "s") == 0)
                    usec *= 1000000;
            } else
                usec *= 1000000;

            INSTR_TIME_SET_CURRENT(now);
            st->until = INSTR_TIME_GET_MICROSEC(now) + usec;
            st->sleeping = 1;

            st->listen = 1;
        } else if (pg_strcasecmp(argv[0], "setshell") == 0) {
            bool ret = runShellCommand(st, argv[1], argv + 2, argc - 2);

            if (timer_exceeded) /* timeout */
                return clientDone(st, true);
            else if (!ret) /* on error */
            {
                st->ecnt++;
                return true;
            } else /* succeeded */
                st->listen = 1;
        } else if (pg_strcasecmp(argv[0], "shell") == 0) {
            bool ret = runShellCommand(st, NULL, argv + 1, argc - 1);

            if (timer_exceeded) /* timeout */
                return clientDone(st, true);
            else if (!ret) /* on error */
            {
                st->ecnt++;
                return true;
            } else /* succeeded */
                st->listen = 1;
        }
        goto top;
    }

    return true;
}

/* discard connections */
static void disconnect_all(CState* state, int length)
{
    int i;

    for (i = 0; i < length; i++) {
        if (state[i].con) {
            PQfinish(state[i].con);
            state[i].con = NULL;
        }
    }
}

/* create tables and setup data */
static void init(void)
{
    /*
     * Note: TPC-B requires at least 100 bytes per row, and the "filler"
     * fields in these table declarations were intended to comply with that.
     * But because they default to NULLs, they don't actually take any space.
     * We could fix that by giving them non-null default values. However, that
     * would completely break comparability of pgbench results with prior
     * versions.  Since pgbench has never pretended to be fully TPC-B
     * compliant anyway, we stick with the historical behavior.
     */
    struct ddlinfo {
        char* table;
        char* cols;
        int declare_fillfactor;
#ifdef PGXC
        char* distribute_by;
#endif
    };
    struct ddlinfo DDLs[] = {{"pgbench_branches",
                                 "bid int not null,bbalance int,filler char(88)",
                                 1
#ifdef PGXC
                                 ,
                                 "distribute by hash (bid)"
#endif
                             },
        {"pgbench_tellers",
            "ttid int not null,bid int,tbalance int,filler char(84)",
            1
#ifdef PGXC
            ,
            "distribute by hash (bid)"
#endif
        },
        {"pgbench_accounts",
            "aid int not null,bid int,abalance int,filler char(84)",
            1
#ifdef PGXC
            ,
            "distribute by hash (bid)"
#endif
        },
        {"pgbench_history",
            "ttid int,bid int,aid int,delta int,mtime timestamp,filler char(22)",
            0
#ifdef PGXC
            ,
            "distribute by hash (bid)"
#endif
        }};

    static char* DDLAFTERs[] = {"alter table pgbench_branches add primary key (bid)",
        "alter table pgbench_tellers add primary key (ttid)",
        "alter table pgbench_accounts add primary key (aid)"};

    static char* DDLAFTERs_mot[] = {"alter foreign table pgbench_branches add primary key (bid)",
        "alter foreign table pgbench_tellers add primary key (ttid)",
        "alter foreign table pgbench_accounts add primary key (aid)"};

#ifdef PGXC
    static char* DDLAFTERs_bid[] = {"alter table pgbench_branches add primary key (bid)",
        "alter table pgbench_tellers add primary key (ttid,bid)",
        "alter table pgbench_accounts add primary key (aid,bid)"};
#endif

    PGconn* con = NULL;
    PGresult* res = NULL;
    char sql[256];
    int i;
    long ttl;

    if ((con = doConnect()) == NULL)
        exit(1);

    for (i = 0; i < (int)lengthof(DDLs); i++) {
        int hasWithOpts = 0;
        char opts[256];
        char check_buffer[512];
        char buffer[512];
        struct ddlinfo* ddl = &DDLs[i];

        /* Remove old table, if it exists. */
        snprintf(check_buffer,
            512,
            "select table_type from information_schema.tables where table_name = '%s' and table_type = 'FOREIGN TABLE'",
            ddl->table);
        PGresult *res = PQexec(con, check_buffer);

        if (PQntuples(res) == 0) {
            snprintf(buffer, 512, "drop table if exists %s", ddl->table);
        } else {
            snprintf(buffer, 512, "drop foreign table if exists %s", ddl->table);
        }
        PQclear(res);
        executeStatement(con, buffer);

        /* Construct new create table statement. */
        opts[0] = '\0';
        hasWithOpts = 0;
        if (with_options || ddl->declare_fillfactor) {
            /* header */
            snprintf(opts + strlen(opts), 256 - strlen(opts), " with (");

            /* only row orientation support fillfactor */
            if ((!orient || strcmp(orient, "row") == 0) && ddl->declare_fillfactor) {
                snprintf(opts + strlen(opts), 256 - strlen(opts), " fillfactor=%d", fillfactor);
                ++hasWithOpts;
            }

            if (with_options) {
                if (hasWithOpts) {
                    snprintf(opts + strlen(opts), 256 - strlen(opts), ",");
                }
                snprintf(opts + strlen(opts), 256 - strlen(opts), " %s ", with_options);
                ++hasWithOpts;
            }

            /* tail */
            snprintf(opts + strlen(opts), 256 - strlen(opts), " )");
        }

        if (tablespace != NULL) {
            char* escape_tablespace = NULL;

            escape_tablespace = PQescapeIdentifier(con, tablespace, strlen(tablespace));
            snprintf(opts + strlen(opts), 256 - strlen(opts), " tablespace %s", escape_tablespace);
            PQfreemem(escape_tablespace);
        }
#ifdef PGXC
        /* Add distribution columns if necessary */
        if (use_branch)
            snprintf(buffer,
                512,
                "create%s table %s(%s)%s %s",
                unlogged_tables ? " unlogged" : "",
                ddl->table,
                ddl->cols,
                opts,
                ddl->distribute_by);
        else
#endif
            if (!is_mot) {
                snprintf(buffer,
                    512,
                    "create%s table %s(%s)%s",
                    unlogged_tables ? " unlogged" : "",
                    ddl->table,
                    ddl->cols,
                    opts);
            } else {
                snprintf(buffer,
                    512,
                    "create foreign table %s(%s)",
                    ddl->table,
                    ddl->cols);
            }

        executeStatement(con, buffer);
    }

#define MINI_BATCH 5000
    /* if mot create primary keys before data load */
    if (is_mot) {
        fprintf(stderr, "set primary key on memory tables...\n");
        for (i = 0; i < (int)lengthof(DDLAFTERs_mot); i++) {
            char buffer[256];
            strncpy(buffer, DDLAFTERs_mot[i], 256);
            executeStatement(con, buffer, true);
        }
    }

    for (i = 0; i < nbranches * scale;) {
        executeStatement(con, "start transaction");
        int k = 0;
        while (k < MINI_BATCH && i < nbranches * scale) {
            snprintf(sql, 256, "insert into pgbench_branches(bid,bbalance) values(%d,0)", i + 1);
            executeStatement(con, sql);
            i++;
            k++;
        }
        executeStatement(con, "commit");
    }

    for (i = 0; i < ntellers * scale;) {
        executeStatement(con, "start transaction");
        int k = 0;
        while (k < MINI_BATCH && i < ntellers * scale) {
            snprintf(
                sql, 256, "insert into pgbench_tellers(ttid,bid,tbalance) values (%d,%d,0)", i + 1, i / ntellers + 1);
            executeStatement(con, sql);
            i++;
            k++;
        }

        executeStatement(con, "commit");
    }

    /*
     * fill the pgbench_accounts table with some data
     */
    fprintf(stderr, "creating tables...\n");

    executeStatement(con, "start transaction");

    executeStatement(con, "truncate pgbench_accounts");

    res = PQexec(con, "copy pgbench_accounts from stdin");
    if (PQresultStatus(res) != PGRES_COPY_IN) {
        fprintf(stderr, "%s", PQerrorMessage(con));
        exit(1);
    }
    PQclear(res);

    ttl = (naccounts * scale) >> 4;
    for (i = 0; i < naccounts * scale; i++) {
        int j = i + 1;

        snprintf(sql, 256, "%d\t%d\t%d\t\n", j, i / naccounts + 1, 0);
        if (PQputline(con, sql)) {
            fprintf(stderr, "PQputline failed\n");
            exit(1);
        }

        if (j % ttl == 0)
            fprintf(stderr, "%d tuples done.\n", j);
    }
    if (PQputline(con, "\\.\n")) {
        fprintf(stderr, "very last PQputline failed\n");
        exit(1);
    }
    if (PQendcopy(con)) {
        fprintf(stderr, "PQendcopy failed\n");
        exit(1);
    }
    executeStatement(con, "commit");

    /*
     * create indexes
     */
    if (!is_mot) {
        fprintf(stderr, "set primary key...\n");
#ifdef PGXC
        /*
         * If all the tables are distributed according to bid, create an index on it
         * instead.
         */
        if (use_branch) {
            for (i = 0; i < (int)lengthof(DDLAFTERs_bid); i++) {
                char buffer[256] = {0};
                errno_t sc_rc = strncpy_s(buffer, sizeof(buffer), DDLAFTERs_bid[i], strlen(DDLAFTERs[i]));
                securec_check(sc_rc, "\0", "\0");

                if (index_tablespace != NULL) {
                    char* escape_tablespace = NULL;

                    escape_tablespace = PQescapeIdentifier(con, index_tablespace, strlen(index_tablespace));
                    snprintf(
                        buffer + strlen(buffer), 256 - strlen(buffer), " using index tablespace %s", escape_tablespace);
                    PQfreemem(escape_tablespace);
                }

                executeStatement(con, buffer, true);
            }
        } else
#endif
            for (i = 0; i < (int)lengthof(DDLAFTERs); i++) {
                char buffer[256] = {0};

                errno_t sc_rc = strncpy_s(buffer, sizeof(buffer), DDLAFTERs[i], strlen(DDLAFTERs[i]));
                securec_check(sc_rc, "\0", "\0");

                if (index_tablespace != NULL) {
                    char* escape_tablespace1 = NULL;

                    escape_tablespace1 = PQescapeIdentifier(con, index_tablespace, strlen(index_tablespace));
                    snprintf(buffer + strlen(buffer),
                        256 - strlen(buffer),
                        " using index tablespace %s",
                        escape_tablespace1);
                    PQfreemem(escape_tablespace1);
                }

                executeStatement(con, buffer, true);
            }
    }

    /* vacuum */
    fprintf(stderr, "vacuum...");
    executeStatement(con, "vacuum analyze pgbench_branches");
    executeStatement(con, "vacuum analyze pgbench_tellers");
    executeStatement(con, "vacuum analyze pgbench_accounts");
    executeStatement(con, "vacuum analyze pgbench_history");

    fprintf(stderr, "done.\n");
    PQfinish(con);
}

/*
 * Parse the raw sql and replace :param to $n.
 */
static bool parseQuery(Command* cmd, const char* raw_sql)
{
    char *sql, *p;

    sql = xstrdup(raw_sql);
    cmd->argc = 1;

    p = sql;
    while ((p = strchr(p, ':')) != NULL) {
        char var[12];
        char* name = NULL;
        int eaten;

        name = parseVariable(p, &eaten);
        if (name == NULL) {
            while (*p == ':') {
                p++;
            }
            continue;
        }

        if (cmd->argc >= MAX_ARGS) {
            fprintf(stderr, "statement has too many arguments (maximum is %d): %s\n", MAX_ARGS - 1, raw_sql);
            return false;
        }

        sprintf(var, "$%d", cmd->argc);
        p = replaceVariable(&sql, p, eaten, var);

        cmd->argv[cmd->argc] = name;
        cmd->argc++;
    }

    cmd->argv[0] = sql;
    return true;
}

/* Parse a command; return a Command struct, or NULL if it's a comment */
static Command* process_commands(char* buf)
{
    const char delim[] = " \f\n\r\t\v";

    Command* my_commands = NULL;
    int j;
    char *p = NULL;;
    char *tok = NULL;

    /* Make the string buf end at the next newline */
    if ((p = strchr(buf, '\n')) != NULL)
        *p = '\0';

    /* Skip leading whitespace */
    p = buf;
    while (isspace((unsigned char)*p)) {
        p++;
    }

    /* If the line is empty or actually a comment, we're done */
    if (*p == '\0' || strncmp(p, "--", 2) == 0)
        return NULL;

    /* Allocate and initialize Command structure */
    my_commands = (Command*)xmalloc(sizeof(Command));
    my_commands->line = xstrdup(buf);
    my_commands->command_num = num_commands++;
    my_commands->type = 0; /* until set */
    my_commands->argc = 0;

    if (*p == '\\') {
        my_commands->type = META_COMMAND;

        j = 0;
        tok = strtok(++p, delim);

        while (tok != NULL) {
            my_commands->argv[j++] = xstrdup(tok);
            my_commands->argc++;
            tok = strtok(NULL, delim);
        }

        if (pg_strcasecmp(my_commands->argv[0], "setrandom") == 0) {
            if (my_commands->argc < 4) {
                fprintf(stderr, "%s: missing argument\n", my_commands->argv[0]);
                exit(1);
            }

            for (j = 4; j < my_commands->argc; j++)
                fprintf(stderr, "%s: extra argument \"%s\" ignored\n", my_commands->argv[0], my_commands->argv[j]);
        } else if (pg_strcasecmp(my_commands->argv[0], "set") == 0) {
            if (my_commands->argc < 3) {
                fprintf(stderr, "%s: missing argument\n", my_commands->argv[0]);
                exit(1);
            }

            for (j = ((my_commands->argc < 5) ? 3 : 5); j < my_commands->argc; j++)
                fprintf(stderr, "%s: extra argument \"%s\" ignored\n", my_commands->argv[0], my_commands->argv[j]);
        } else if (pg_strcasecmp(my_commands->argv[0], "sleep") == 0) {
            if (my_commands->argc < 2) {
                fprintf(stderr, "%s: missing argument\n", my_commands->argv[0]);
                exit(1);
            }

            /*
             * Split argument into number and unit to allow "sleep 1ms" etc.
             * We don't have to terminate the number argument with null
             * because it will be parsed with atoi, which ignores trailing
             * non-digit characters.
             */
            if (my_commands->argv[1][0] != ':') {
                char* c = my_commands->argv[1];

                while (isdigit((unsigned char)*c)) {
                    c++;
                }
                if (*c) {
                    my_commands->argv[2] = c;
                    if (my_commands->argc < 3)
                        my_commands->argc = 3;
                }
            }

            if (my_commands->argc >= 3) {
                if (pg_strcasecmp(my_commands->argv[2], "us") != 0 && pg_strcasecmp(my_commands->argv[2], "ms") != 0 &&
                    pg_strcasecmp(my_commands->argv[2], "s") != 0) {
                    fprintf(stderr,
                        "%s: unknown time unit '%s' - must be us, ms or s\n",
                        my_commands->argv[0],
                        my_commands->argv[2]);
                    exit(1);
                }
            }

            for (j = 3; j < my_commands->argc; j++)
                fprintf(stderr, "%s: extra argument \"%s\" ignored\n", my_commands->argv[0], my_commands->argv[j]);
        } else if (pg_strcasecmp(my_commands->argv[0], "setshell") == 0) {
            if (my_commands->argc < 3) {
                fprintf(stderr, "%s: missing argument\n", my_commands->argv[0]);
                exit(1);
            }
        } else if (pg_strcasecmp(my_commands->argv[0], "shell") == 0) {
            if (my_commands->argc < 1) {
                fprintf(stderr, "%s: missing command\n", my_commands->argv[0]);
                exit(1);
            }
        } else {
            fprintf(stderr, "Invalid command %s\n", my_commands->argv[0]);
            exit(1);
        }
    } else {
        my_commands->type = SQL_COMMAND;

        switch (querymode) {
            case QUERY_SIMPLE:
                my_commands->argv[0] = xstrdup(p);
                my_commands->argc++;
                break;
            case QUERY_EXTENDED:
            case QUERY_PREPARED:
                if (!parseQuery(my_commands, p))
                    exit(1);
                break;
            default:
                exit(1);
        }
    }

    return my_commands;
}

static int process_file(char* filename)
{
#define COMMANDS_ALLOC_NUM 128

    Command** my_commands;
    FILE* fd = NULL;
    int lineno;
    char buf[BUFSIZ * 8];
    int alloc_num;

    if (num_files >= MAX_FILES) {
        fprintf(stderr, "Up to only %d SQL files are allowed\n", MAX_FILES);
        exit(1);
    }

    alloc_num = COMMANDS_ALLOC_NUM;
    my_commands = (Command**)xmalloc(sizeof(Command*) * alloc_num);

    if (strcmp(filename, "-") == 0)
        fd = stdin;
    else if ((fd = fopen(filename, "r")) == NULL) {
        fprintf(stderr, "%s: %s\n", filename, strerror(errno));
        return false;
    }

    lineno = 0;

    while (fgets(buf, sizeof(buf), fd) != NULL) {
        Command* command = NULL;

        command = process_commands(buf);
        if (command == NULL)
            continue;

        my_commands[lineno] = command;
        lineno++;

        if (lineno >= alloc_num) {
            alloc_num += COMMANDS_ALLOC_NUM;
            my_commands = (Command**)xrealloc(my_commands, sizeof(Command*) * alloc_num);
        }
    }
    fclose(fd);

    my_commands[lineno] = NULL;

    sql_files[num_files++] = my_commands;

    return true;
}

static Command** process_builtin(char* tb)
{
#define COMMANDS_ALLOC_NUM 128

    Command** my_commands;
    int lineno;
    char buf[BUFSIZ];
    int alloc_num;

    alloc_num = COMMANDS_ALLOC_NUM;
    my_commands = (Command**)xmalloc(sizeof(Command*) * alloc_num);

    lineno = 0;

    for (;;) {
        char* p = NULL;
        Command* command = NULL;

        p = buf;
        while (*tb && *tb != '\n') {
            *p++ = *tb++;
        }

        if (*tb == '\0')
            break;

        if (*tb == '\n')
            tb++;

        *p = '\0';

        command = process_commands(buf);
        if (command == NULL) {
            continue;
        }

        my_commands[lineno] = command;
        lineno++;

        if (lineno >= alloc_num) {
            alloc_num += COMMANDS_ALLOC_NUM;
            my_commands = (Command**)xrealloc(my_commands, sizeof(Command*) * alloc_num);
        }
    }

    my_commands[lineno] = NULL;

    return my_commands;
}

/* print out results */
static void printResults(int ttype, int normal_xacts, int nclients, TState* threads, int nthreads,
    instr_time total_time, instr_time conn_total_time, int64 total_latencies, int64 total_sqlats, int64 throttle_lag,
    int64 throttle_lag_max)
{
    double time_include, tps_include, tps_exclude;
    char* s = NULL;

    time_include = INSTR_TIME_GET_DOUBLE(total_time);
    tps_include = normal_xacts / time_include;
    tps_exclude = normal_xacts / (time_include - (INSTR_TIME_GET_DOUBLE(conn_total_time) / nthreads));

    if (ttype == 0)
        s = "TPC-B (sort of)";
    else if (ttype == 2)
        s = "Update only pgbench_accounts";
    else if (ttype == 1)
        s = "SELECT only";
    else
        s = "Custom query";

    printf("transaction type: %s\n", s);
    printf("scaling factor: %d\n", scale);
    printf("query mode: %s\n", QUERYMODE[querymode]);
    printf("number of clients: %d\n", nclients);
    printf("number of threads: %d\n", nthreads);
    if (duration <= 0) {
        printf("number of transactions per client: %d\n", nxacts);
        printf("number of transactions actually processed: %d/%d\n", normal_xacts, nxacts * nclients);
    } else {
        printf("duration: %d s\n", duration);
        printf("number of transactions actually processed: %d\n", normal_xacts);
    }

    if (throttle_delay || progress) {
        /* compute and show latency average and standard deviation */
        double latency = 0.001 * total_latencies / normal_xacts;
        double sqlat = (double)total_sqlats / normal_xacts;
        printf("latency average: %.3f ms\n"
               "latency stddev: %.3f ms\n",
            latency,
            0.001 * sqrt(sqlat - 1000000.0 * latency * latency));
    } else {
        /* only an average latency computed from the duration is available */
        printf("latency average: %.3f ms\n", 1000.0 * duration * nclients / normal_xacts);
    }

    if (throttle_delay) {
        /*
         * Report average transaction lag under rate limit throttling.  This
         * is the delay between scheduled and actual start times for the
         * transaction.  The measured lag may be caused by thread/client load,
         * the database load, or the Poisson throttling process.
         */
        printf("rate limit schedule lag: avg %.3f (max %.3f) ms\n",
            0.001 * throttle_lag / normal_xacts,
            0.001 * throttle_lag_max);
    }

    printf("tps = %f (including connections establishing)\n", tps_include);
    printf("tps = %f (excluding connections establishing)\n", tps_exclude);

    /* Report per-command latencies */
    if (is_latencies) {
        int i;

        for (i = 0; i < num_files; i++) {
            Command** commands;

            if (num_files > 1)
                printf("statement latencies in milliseconds, file %d:\n", i + 1);
            else
                printf("statement latencies in milliseconds:\n");

            for (commands = sql_files[i]; *commands != NULL; commands++) {
                Command* command = *commands;
                int cnum = command->command_num;
                double total_time;
                instr_time total_exec_elapsed;
                int total_exec_count;
                int t;

                /* Accumulate per-thread data for command */
                INSTR_TIME_SET_ZERO(total_exec_elapsed);
                total_exec_count = 0;
                for (t = 0; t < nthreads; t++) {
                    TState* thread = &threads[t];

                    INSTR_TIME_ADD(total_exec_elapsed, thread->exec_elapsed[cnum]);
                    total_exec_count += thread->exec_count[cnum];
                }

                if (total_exec_count > 0) {
                    total_time = INSTR_TIME_GET_MILLISEC(total_exec_elapsed) / (double)total_exec_count;
                } else {
                    total_time = 0.0;
                }

                printf("\t%f\t%s\n", total_time, command->line);
            }
        }
    }
}

int main(int argc, char** argv)
{
    int c;
    int nclients = 1;           /* default number of simulated clients */
    int nthreads = 1;           /* default number of threads */
    int is_init_mode = 0;       /* initialize mode? */
    int is_no_vacuum = 0;       /* no vacuum at all before testing? */
    int do_vacuum_accounts = 0; /* do vacuum accounts before testing? */
    int ttype = 0;              /* transaction type. 0: TPC-B, 1: SELECT only,
                                 * 2: skip update of branches and tellers */
    int optindex;
    char* filename = NULL;
    bool scale_given = false;

    CState* state = NULL;   /* status of clients */
    TState* threads = NULL; /* array of thread */

    instr_time start_time; /* start up time */
    instr_time total_time;
    instr_time conn_total_time;
    int total_xacts = 0;
    int64 total_latencies = 0;
    int64 total_sqlats = 0;
    int64 throttle_lag = 0;
    int64 throttle_lag_max = 0;

    int i;

    static struct option long_options[] = {{"index-tablespace", required_argument, NULL, 3},
        {"tablespace", required_argument, NULL, 2},
        {"unlogged-tables", no_argument, &unlogged_tables, 1},
        {NULL, 0, NULL, 0}};

#ifdef HAVE_GETRLIMIT
    struct rlimit rlim;
#endif

    PGconn* con = NULL;
    PGresult* res = NULL;
    char* env = NULL;

    char val[64];

    const char* progname = NULL;

    progname = get_progname(argv[0]);

    if (argc > 1) {
        if (strcmp(argv[1], "--help") == 0 || strcmp(argv[1], "-?") == 0) {
            usage(progname);
            exit(0);
        }
        if (strcmp(argv[1], "--version") == 0 || strcmp(argv[1], "-V") == 0) {
            puts("pgbench (PostgreSQL) " PG_VERSION);
            exit(0);
        }
    }

#ifdef WIN32
    /* stderr is buffered on Win32. */
    setvbuf(stderr, NULL, _IONBF, 0);
#endif

    if ((env = getenv("PGHOST")) != NULL && *env != '\0')
        pghost = env;
    if ((env = getenv("PGPORT")) != NULL && *env != '\0')
        pgport = env;
    else if ((env = getenv("PGUSER")) != NULL && *env != '\0')
        login = env;

    state = (CState*)xmalloc(sizeof(CState));
    memset(state, 0, sizeof(CState));

#ifdef PGXC
    while ((c = getopt_long(argc, argv, "ih:mknvp:dSNc:j:Crs:t:T:U:lf:D:F:M:O:P:R:W:", long_options, &optindex)) != -1)
#else
    while ((c = getopt_long(argc, argv, "ih:mnvp:dSNc:j:Crs:t:T:U:lf:D:F:M:P:R:W:", long_options, &optindex)) != -1)
#endif
    {
        switch (c) {
            case 'i':
                is_init_mode++;
                break;
#ifdef PGXC
            case 'k':
                use_branch = true;
                break;
#endif
            case 'm':
                is_mot = true;
                break;
            case 'h':
                pghost = optarg;
                break;
            case 'n':
                is_no_vacuum++;
                break;
            case 'v':
                do_vacuum_accounts++;
                break;
            case 'p':
                pgport = optarg;
                break;
            case 'd':
                debug++;
                break;
            case 'S':
                ttype = 1;
                break;
            case 'N':
                ttype = 2;
                break;
            case 'c':
                nclients = atoi(optarg);

#ifdef HAVE_POLL
                if (nclients <= 0)
#else
                if (nclients <= 0 || nclients > MAXCLIENTS)
#endif
                {
                    fprintf(stderr, "invalid number of clients: %d\n", nclients);
                    exit(1);
                }
#ifdef HAVE_GETRLIMIT
#ifdef RLIMIT_NOFILE /* most platforms use RLIMIT_NOFILE */
                if (getrlimit(RLIMIT_NOFILE, &rlim) == -1)
#else  /* but BSD doesn't ... */
                if (getrlimit(RLIMIT_OFILE, &rlim) == -1)
#endif /* RLIMIT_NOFILE */
                {
                    fprintf(stderr, "getrlimit failed: %s\n", strerror(errno));
                    exit(1);
                }
                if ((int)rlim.rlim_cur <= (nclients + 2)) {
                    fprintf(stderr,
                        "You need at least %d open files but you are only allowed to use %ld.\n",
                        nclients + 2,
                        (long)rlim.rlim_cur);
                    fprintf(stderr, "Use limit/ulimit to increase the limit before using pgbench.\n");
                    exit(1);
                }
#endif /* HAVE_GETRLIMIT */
                break;
            case 'j': /* jobs */
                nthreads = atoi(optarg);
                if (nthreads <= 0) {
                    fprintf(stderr, "invalid number of threads: %d\n", nthreads);
                    exit(1);
                }
                break;
            case 'C':
                is_connect = true;
                break;
            case 'r':
                is_latencies = true;
                break;
            case 's':
                scale_given = true;
                scale = atoi(optarg);
                if (scale <= 0) {
                    fprintf(stderr, "invalid scaling factor: %d\n", scale);
                    exit(1);
                }
                break;
            case 't':
                if (duration > 0) {
                    fprintf(stderr, "specify either a number of transactions (-t) or a duration (-T), not both.\n");
                    exit(1);
                }
                nxacts = atoi(optarg);
                if (nxacts <= 0) {
                    fprintf(stderr, "invalid number of transactions: %d\n", nxacts);
                    exit(1);
                }
                break;
            case 'T':
                if (nxacts > 0) {
                    fprintf(stderr, "specify either a number of transactions (-t) or a duration (-T), not both.\n");
                    exit(1);
                }
                duration = atoi(optarg);
                if (duration <= 0) {
                    fprintf(stderr, "invalid duration: %d\n", duration);
                    exit(1);
                }
                break;
            case 'U':
                login = optarg;
                break;
            case 'W':
                secrete = optarg;
                break;
            case 'l':
                use_log = true;
                break;
            case 'f':
                ttype = 3;
                filename = optarg;
                if (process_file(filename) == false || *sql_files[num_files - 1] == NULL)
                    exit(1);
                break;
            case 'D': {
                char* p = NULL;

                if ((p = strchr(optarg, '=')) == NULL || p == optarg || *(p + 1) == '\0') {
                    fprintf(stderr, "invalid variable definition: %s\n", optarg);
                    exit(1);
                }

                *p++ = '\0';
                if (!putVariable(&state[0], "option", optarg, p))
                    exit(1);
            } break;
            case 'F':
                fillfactor = atoi(optarg);
                if (is_mot) {
                    fprintf(stderr, "fillfactor is not supported with memory tables\n");
                    exit(1);
                }
                if ((fillfactor < 10) || (fillfactor > 100)) {
                    fprintf(stderr, "invalid fillfactor: %d\n", fillfactor);
                    exit(1);
                }
                break;
            case 'M':
                if (num_files > 0) {
                    fprintf(stderr, "query mode (-M) should be specifiled before transaction scripts (-f)\n");
                    exit(1);
                }
                for (int iquerymode = 0; iquerymode < NUM_QUERYMODE; iquerymode++) {
                    querymode = (QueryMode)iquerymode;
                    if (strcmp(optarg, QUERYMODE[querymode]) == 0)
                        break;
                }
                if (querymode >= NUM_QUERYMODE) {
                    fprintf(stderr, "invalid query mode (-M): %s\n", optarg);
                    exit(1);
                }
                break;
#ifdef PGXC
            case 'O':
                orient = xstrdup(optarg);
                break;
#endif

            case 'P':
                progress = atoi(optarg);
                if (progress <= 0) {
                    fprintf(stderr, "thread progress delay (-P) must be positive (%s)\n", optarg);
                    exit(1);
                }
                break;
            case 'R': {
                /* get a double from the beginning of option value */
                double throttle_value = atof(optarg);
                if (throttle_value <= 0.0) {
                    fprintf(stderr, "invalid rate limit: %s\n", optarg);
                    exit(1);
                }
                /* Invert rate limit into a time offset */
                throttle_delay = (int64)(1000000.0 / throttle_value);
            } break;
            case 0:
                /* This covers long options which take no argument. */
                break;
            case 2: /* tablespace */
                if (is_mot) {
                    fprintf(stderr, "tablespace is not supported with memory tables\n");
                    exit(1);
                }
                tablespace = optarg;
                break;
            case 3: /* index-tablespace */
                if (is_mot) {
                    fprintf(stderr, "index_tablespace is not supported with memory tables\n");
                    exit(1);
                }
                index_tablespace = optarg;
                break;
            default:
                fprintf(stderr, _("Try \"%s --help\" for more information.\n"), progname);
                exit(1);
                break;
        }
    }
    /* compute a per thread delay */
    throttle_delay *= nthreads;

    if (orient) {
#define MIN_LEN(s, x) (((strlen(s) > x) ? x : strlen(s)))
        if (strncmp(orient, "row", MIN_LEN(orient, 3)) == 0) {
            with_options = xstrdup("orientation = row");
        } else if (strncmp(orient, "column", MIN_LEN(orient, 6)) == 0) {
            with_options = xstrdup("orientation = column");
        } else if (strncmp(orient, "orc", MIN_LEN(orient, 3)) == 0) {
            with_options = xstrdup("orientation = orc, version=0.12");
        } else {
            fprintf(
                stderr, "Unknown orientation option: %s.\nTry \"%s --help\" for more information.\n", orient, progname);
            exit(1);
        }
    }

    if (argc > optind)
        dbName = argv[optind];
    else {
        if ((env = getenv("PGDATABASE")) != NULL && *env != '\0') {
            dbName = env;
        } else if (login != NULL && *login != '\0') {
            dbName = login;
        } else {
            dbName = "";
        }
    }

    if (is_init_mode) {
        init();

        if (orient) {
            free(orient);
            orient = NULL;
        }
        if (with_options) {
            free(with_options);
            with_options = NULL;
        }

        exit(0);
    }

    /* Use DEFAULT_NXACTS if neither nxacts nor duration is specified. */
    if (nxacts <= 0 && duration <= 0)
        nxacts = DEFAULT_NXACTS;

    if (nclients % nthreads != 0) {
        fprintf(stderr, "number of clients (%d) must be a multiple of number of threads (%d)\n", nclients, nthreads);
        exit(1);
    }

    /*
     * is_latencies only works with multiple threads in thread-based
     * implementations, not fork-based ones, because it supposes that the
     * parent can see changes made to the per-thread execution stats by child
     * threads.  It seems useful enough to accept despite this limitation, but
     * perhaps we should FIXME someday (by passing the stats data back up
     * through the parent-to-child pipes).
     */
#ifndef ENABLE_THREAD_SAFETY
    if (is_latencies && nthreads > 1) {
        fprintf(stderr, "-r does not work with -j larger than 1 on this platform.\n");
        exit(1);
    }
#endif

    // Prepared protocol only prepares the statement once so reconnect for each
    // query submission won't work.
    //
    if (querymode == QUERY_PREPARED && is_connect) {
        fprintf(stderr, "-C does not work with -M prepared protocol.\n");
        exit(1);
    }

    /*
     * save main process id in the global variable because process id will be
     * changed after fork.
     */
    main_pid = (int)getpid();
    progress_nclients = nclients;
    progress_nthreads = nthreads;

    if (nclients > 1) {
        state = (CState*)xrealloc(state, sizeof(CState) * nclients);
        memset(state + 1, 0, sizeof(CState) * (nclients - 1));

        /* copy any -D switch values to all clients */
        for (i = 1; i < nclients; i++) {
            int j;

            state[i].id = i;
            for (j = 0; j < state[0].nvariables; j++) {
                if (!putVariable(&state[i], "startup", state[0].variables[j].name, state[0].variables[j].value))
                    exit(1);
            }
        }
    }

    if (debug) {
        if (duration <= 0)
            printf(
                "pghost: %s pgport: %s nclients: %d nxacts: %d dbName: %s\n", pghost, pgport, nclients, nxacts, dbName);
        else
            printf("pghost: %s pgport: %s nclients: %d duration: %d dbName: %s\n",
                pghost,
                pgport,
                nclients,
                duration,
                dbName);
    }

    /* opening connection... */
    con = doConnect();
    if (con == NULL)
        exit(1);

    if (PQstatus(con) == CONNECTION_BAD) {
        fprintf(stderr, "Connection to database '%s' failed.\n", dbName);
        fprintf(stderr, "%s", PQerrorMessage(con));
        exit(1);
    }

    if (ttype != 3) {
        /*
         * get the scaling factor that should be same as count(*) from
         * pgbench_branches if this is not a custom query
         */
        res = PQexec(con, "select count(*) from pgbench_branches");
        if (PQresultStatus(res) != PGRES_TUPLES_OK) {
            fprintf(stderr, "%s", PQerrorMessage(con));
            exit(1);
        }
        scale = atoi(PQgetvalue(res, 0, 0));
        if (scale < 0) {
            fprintf(stderr, "count(*) from pgbench_branches invalid (%d)\n", scale);
            exit(1);
        }
        PQclear(res);

        /* warn if we override user-given -s switch */
        if (scale_given)
            fprintf(stderr, "Scale option ignored, using pgbench_branches table count = %d\n", scale);
    }

    /*
     * :scale variables normally get -s or database scale, but don't override
     * an explicit -D switch
     */
    if (getVariable(&state[0], "scale") == NULL) {
        snprintf(val, sizeof(val), "%d", scale);
        for (i = 0; i < nclients; i++) {
            if (!putVariable(&state[i], "startup", "scale", val))
                exit(1);
        }
    }

    if (!is_no_vacuum) {
        fprintf(stderr, "starting vacuum...");
        executeStatement(con, "vacuum pgbench_branches");
        executeStatement(con, "vacuum pgbench_tellers");
        executeStatement(con, "truncate pgbench_history");
        fprintf(stderr, "end.\n");

        if (do_vacuum_accounts) {
            fprintf(stderr, "starting vacuum pgbench_accounts...");
            executeStatement(con, "vacuum analyze pgbench_accounts");
            fprintf(stderr, "end.\n");
        }
    }
    PQfinish(con);

    /* set random seed */
    INSTR_TIME_SET_CURRENT(start_time);
    srandom((unsigned int)INSTR_TIME_GET_MICROSEC(start_time));

    /* process builtin SQL scripts */
    switch (ttype) {
        case 0:
#ifdef PGXC
            if (use_branch)
                sql_files[0] = process_builtin(tpc_b_bid);
            else
#endif
                sql_files[0] = process_builtin(tpc_b);
            num_files = 1;
            break;

        case 1:
            sql_files[0] = process_builtin(select_only);
            num_files = 1;
            break;

        case 2:
#ifdef PGXC
            if (use_branch)
                sql_files[0] = process_builtin(simple_update_bid);
            else
#endif
                sql_files[0] = process_builtin(simple_update);
            num_files = 1;
            break;

        default:
            break;
    }

    /* set up thread data structures */
    threads = (TState*)xmalloc(sizeof(TState) * nthreads);
    for (i = 0; i < nthreads; i++) {
        TState* thread = &threads[i];

        thread->tid = i;
        thread->state = &state[nclients / nthreads * i];
        thread->nstate = nclients / nthreads;
        thread->random_state[0] = random();
        thread->random_state[1] = random();
        thread->random_state[2] = random();

        if (is_latencies) {
            /* Reserve memory for the thread to store per-command latencies */
            int t;

            thread->exec_elapsed = (instr_time*)xmalloc(sizeof(instr_time) * num_commands);
            thread->exec_count = (int*)xmalloc(sizeof(int) * num_commands);

            for (t = 0; t < num_commands; t++) {
                INSTR_TIME_SET_ZERO(thread->exec_elapsed[t]);
                thread->exec_count[t] = 0;
            }
        } else {
            thread->exec_elapsed = NULL;
            thread->exec_count = NULL;
        }
    }

    /* get start up time */
    INSTR_TIME_SET_CURRENT(start_time);

    /* set alarm if duration is specified. */
    if (duration > 0)
        setalarm(duration);

    /* start threads */
    for (i = 0; i < nthreads; i++) {
        TState* thread = &threads[i];

        INSTR_TIME_SET_CURRENT(thread->start_time);

        /* the first thread (i = 0) is executed by main thread */
        if (i > 0) {
            int err = pthread_create(&thread->thread, NULL, threadRun, thread);

            if (err != 0 || thread->thread == INVALID_THREAD) {
                fprintf(stderr, "cannot create thread: %s\n", strerror(err));
                exit(1);
            }
        } else {
            thread->thread = INVALID_THREAD;
        }
    }

    /* wait for threads and accumulate results */
    total_xacts = 0;
    INSTR_TIME_SET_ZERO(conn_total_time);
    for (i = 0; i < nthreads; i++) {
        void* ret = NULL;

        if (threads[i].thread == INVALID_THREAD)
            ret = threadRun(&threads[i]);
        else
            pthread_join(threads[i].thread, &ret);

        if (ret != NULL) {
            TResult* r = (TResult*)ret;

            total_xacts += r->xacts;
            total_latencies += r->latencies;
            total_sqlats += r->sqlats;
            throttle_lag += r->throttle_lag;
            if (r->throttle_lag_max > throttle_lag_max)
                throttle_lag_max = r->throttle_lag_max;
            INSTR_TIME_ADD(conn_total_time, r->conn_time);
            free(ret);
        }
    }
    disconnect_all(state, nclients);

    /* get end time */
    INSTR_TIME_SET_CURRENT(total_time);
    INSTR_TIME_SUBTRACT(total_time, start_time);
    printResults(ttype,
        total_xacts,
        nclients,
        threads,
        nthreads,
        total_time,
        conn_total_time,
        total_latencies,
        total_sqlats,
        throttle_lag,
        throttle_lag_max);

    return 0;
}

static void* threadRun(void* arg)
{
    TState* thread = (TState*)arg;
    CState* state = thread->state;
    TResult* result = NULL;
    FILE* logfile = NULL; /* per-thread log file */
    instr_time start, end;
    int nstate = thread->nstate;
    int remains = nstate; /* number of remaining clients */
    int i, j;

    /* for reporting progress: */
    int64 thread_start = INSTR_TIME_GET_MICROSEC(thread->start_time);
    int64 last_report = thread_start;
    int64 next_report = last_report + (int64)progress * 1000000;
    int64 last_count = 0, last_lats = 0, last_sqlats = 0, last_lags = 0;

#ifdef HAVE_POLL
    struct pollfd* ufds = (pollfd*)xmalloc(nstate * sizeof(pollfd)); /*nstate = nclients /nthreads */
    int nfds = 0;                                                    /*the count number of fd */
#else
    fd_set input_mask;
    int maxsock = -1; /* max socket number to be waited, initialized as -1 */
#endif
    /*
     * Initialize throttling rate target for all of the thread's clients.  It
     * might be a little more accurate to reset thread->start_time here too.
     * The possible drift seems too small relative to typical throttle delay
     * times to worry about it.
     */
    INSTR_TIME_SET_CURRENT(start);
    thread->throttle_trigger = INSTR_TIME_GET_MICROSEC(start);
    thread->throttle_lag = 0;
    thread->throttle_lag_max = 0;

    result = (TResult*)xmalloc(sizeof(TResult));
    INSTR_TIME_SET_ZERO(result->conn_time);

    /* open log file if requested */
    if (use_log) {
        char logpath[64];

        if (thread->tid == 0)
            snprintf(logpath, sizeof(logpath), "pgbench_log.%d", main_pid);
        else
            snprintf(logpath, sizeof(logpath), "pgbench_log.%d.%d", main_pid, thread->tid);
        logfile = fopen(logpath, "w");

        if (logfile == NULL) {
            fprintf(stderr, "Couldn't open logfile \"%s\": %s", logpath, strerror(errno));
            goto done;
        }
    }

    if (!is_connect) {
        /* make connections to the database */
        for (i = 0; i < nstate; i++) {
            if ((state[i].con = doConnect()) == NULL)
                goto done;
        }
    }

    /* time after thread and connections set up */
    INSTR_TIME_SET_CURRENT(result->conn_time);
    INSTR_TIME_SUBTRACT(result->conn_time, thread->start_time);

    /* send start up queries in async manner */
    for (i = 0; i < nstate; i++) {
        CState* st = &state[i];
        Command** commands = sql_files[st->use_file];
        int prev_ecnt = st->ecnt;

        st->use_file = getrand(thread, 0, num_files - 1);
        if (!doCustom(thread, st, &result->conn_time, logfile))
            remains--; /* I've aborted */

        if (st->ecnt > prev_ecnt && commands[st->state]->type == META_COMMAND) {
            fprintf(stderr, "Client %d aborted in state %d. Execution meta-command failed.\n", i, st->state);
            remains--; /* I've aborted */
            PQfinish(st->con);
            st->con = NULL;
        }
    }

    while (remains > 0) {
#ifdef HAVE_POLL
        nfds = 0; /*the count number of fd */
        memset(ufds, 0, nstate * sizeof(pollfd));
#else
        FD_ZERO(&input_mask);
        maxsock = -1; /* max socket number to be waited, initialized as -1 */
#endif

        int64 now_usec = 0;
        int64 min_usec = INT64_MAX;

        for (i = 0; i < nstate; i++) {
            CState* st = &state[i];
            Command** commands = sql_files[st->use_file];
            int sock;

            if (st->con == NULL) {
                continue;
            } else if (st->sleeping) {
                if (st->throttling && timer_exceeded) {
                    /* interrupt client which has not started a transaction */
                    remains--;
                    st->sleeping = 0;
                    st->throttling = false;
                    PQfinish(st->con);
                    st->con = NULL;
                    continue;
                } else /* just a nap from the script */
                {
                    int this_usec;

                    if (min_usec == INT64_MAX) {
                        instr_time now;

                        INSTR_TIME_SET_CURRENT(now);
                        now_usec = INSTR_TIME_GET_MICROSEC(now);
                    }

                    this_usec = st->until - now_usec;
                    if (min_usec > this_usec)
                        min_usec = this_usec;
                }
            } else if (commands[st->state]->type == META_COMMAND) {
                min_usec = 0; /* the connection is ready to run */
                break;
            }

            sock = PQsocket(st->con);
            if (sock < 0) {
                fprintf(stderr, "bad socket: %s\n", strerror(errno));
                goto done;
            }
#ifdef HAVE_POLL
            ufds[nfds].fd = sock;
            ufds[nfds].events = POLLIN | POLLPRI | POLLRDHUP | POLLERR | POLLHUP;
            ++nfds;
#else
            FD_SET(sock, &input_mask);

            if (maxsock < sock)
                maxsock = sock;
#endif
        }

#ifdef HAVE_POLL
        if (min_usec > 0 && nfds > 0) {
            int nsocks; /* return from select(2) */

            if (min_usec != INT64_MAX) {
                nsocks = poll(ufds, nfds, min_usec / 1000);
            } else {
                nsocks = poll(ufds, nfds, -1); /*wait for ever, until some events happend in the sockets*/
            }

            if (nsocks < 0) {
                if (errno == EINTR)
                    continue;
                /* must be something wrong */
                fprintf(stderr, "poll failed: %s\n", strerror(errno));
                goto done;
            }
        }

        for (i = 0; i < nfds; i++) {
            CState* st = NULL;

            for (j = 0; j < nstate; j++) {
                st = &state[j];
                if (ufds[i].fd == PQsocket(st->con))
                    break;
            }

            Command** commands = sql_files[st->use_file];
            int prev_ecnt = st->ecnt;
            if (st->con &&
                (ufds[i].revents & (POLLIN | POLLPRI | POLLHUP) || commands[st->state]->type == META_COMMAND))

            {

                if (!doCustom(thread, st, &result->conn_time, logfile))
                    remains--; /* I've aborted */

                ufds[i].revents = 0;
            }

            if (st->ecnt > prev_ecnt && commands[st->state]->type == META_COMMAND) {
                fprintf(stderr, "Client %d aborted in state %d. Execution of meta-command failed.\n", i, st->state);
                remains--; /* I've aborted */
                PQfinish(st->con);
                st->con = NULL;
            }
        }
#else
        if (min_usec > 0 && maxsock != -1) {
            int nsocks; /* return from select(2) */

            if (min_usec != INT64_MAX) {
                struct timeval timeout;

                timeout.tv_sec = min_usec / 1000000;
                timeout.tv_usec = min_usec % 1000000;
                nsocks = select(maxsock + 1, &input_mask, NULL, NULL, &timeout);
            } else {
                nsocks = select(maxsock + 1, &input_mask, NULL, NULL, NULL);
            }

            if (nsocks < 0) {
                if (errno == EINTR)
                    continue;
                /* must be something wrong */
                fprintf(stderr, "select failed: %s\n", strerror(errno));
                goto done;
            }
        }

        /* ok, backend returns reply */
        for (i = 0; i < nstate; i++) {
            CState* st = &state[i];
            Command** commands = sql_files[st->use_file];
            int prev_ecnt = st->ecnt;
            if (st->con && (FD_ISSET(PQsocket(st->con), &input_mask) || commands[st->state]->type == META_COMMAND)) {

                if (!doCustom(thread, st, &result->conn_time, logfile))
                    remains--; /* I've aborted */
            }

            if (st->ecnt > prev_ecnt && commands[st->state]->type == META_COMMAND) {
                fprintf(stderr, "Client %d aborted in state %d. Execution of meta-command failed.\n", i, st->state);
                remains--; /* I've aborted */
                PQfinish(st->con);
                st->con = NULL;
            }
        }
#endif

#ifdef PTHREAD_FORK_EMULATION
        /* each process reports its own progression */
        if (progress) {
            instr_time now_time;
            int64 now;
            INSTR_TIME_SET_CURRENT(now_time);
            now = INSTR_TIME_GET_MICROSEC(now_time);
            if (now >= next_report) {
                /* generate and show report */
                int64 count = 0, lats = 0, sqlats = 0;
                int64 lags = thread->throttle_lag;
                int64 run = now - last_report;
                double tps, total_run, latency, sqlat, stdev, lag;

                for (i = 0; i < nstate; i++) {
                    count += state[i].cnt;
                    lats += state[i].txn_latencies;
                    sqlats += state[i].txn_sqlats;
                }

                total_run = (now - thread_start) / 1000000.0;
                tps = 1000000.0 * (count - last_count) / run;
                latency = 0.001 * (lats - last_lats) / (count - last_count);
                sqlat = 1.0 * (sqlats - last_sqlats) / (count - last_count);
                stdev = 0.001 * sqrt(sqlat - 1000000.0 * latency * latency);
                lag = 0.001 * (lags - last_lags) / (count - last_count);

                if (throttle_delay)
                    fprintf(stderr,
                        "progress %d: %.1f s, %.1f tps, "
                        "lat %.3f ms stddev %.3f, lag %.3f ms\n",
                        thread->tid,
                        total_run,
                        tps,
                        latency,
                        stdev,
                        lag);
                else
                    fprintf(stderr,
                        "progress %d: %.1f s, %.1f tps, "
                        "lat %.3f ms stddev %.3f\n",
                        thread->tid,
                        total_run,
                        tps,
                        latency,
                        stdev);

                last_count = count;
                last_lats = lats;
                last_sqlats = sqlats;
                last_lags = lags;
                last_report = now;
                next_report += progress * 1000000;
            }
        }
#else
        /* progress report by thread 0 for all threads */
        if (progress && thread->tid == 0) {
            instr_time now_time;
            int64 now;
            INSTR_TIME_SET_CURRENT(now_time);
            now = INSTR_TIME_GET_MICROSEC(now_time);
            if (now >= next_report) {
                /* generate and show report */
                int64 count = 0, lats = 0, sqlats = 0, lags = 0;
                int64 run = now - last_report;
                double tps, total_run, latency, sqlat, lag, stdev;

                for (i = 0; i < progress_nclients; i++) {
                    count += state[i].cnt;
                    lats += state[i].txn_latencies;
                    sqlats += state[i].txn_sqlats;
                }

                for (i = 0; i < progress_nthreads; i++)
                    lags += thread[i].throttle_lag;

                total_run = (now - thread_start) / 1000000.0;
                tps = 1000000.0 * (count - last_count) / run;
                latency = 0.001 * (lats - last_lats) / (count - last_count);
                sqlat = 1.0 * (sqlats - last_sqlats) / (count - last_count);
                stdev = 0.001 * sqrt(sqlat - 1000000.0 * latency * latency);
                lag = 0.001 * (lags - last_lags) / (count - last_count);

                if (throttle_delay)
                    fprintf(stderr,
                        "progress: %.1f s, %.1f tps, "
                        "lat %.3f ms stddev %.3f, lag %.3f ms\n",
                        total_run,
                        tps,
                        latency,
                        stdev,
                        lag);
                else
                    fprintf(stderr,
                        "progress: %.1f s, %.1f tps, "
                        "lat %.3f ms stddev %.3f\n",
                        total_run,
                        tps,
                        latency,
                        stdev);

                last_count = count;
                last_lats = lats;
                last_sqlats = sqlats;
                last_lags = lags;
                last_report = now;
                next_report += progress * 1000000;
            }
        }
#endif /* PTHREAD_FORK_EMULATION */
    }

done:
#ifdef HAVE_POLL
    free(ufds);
#endif
    INSTR_TIME_SET_CURRENT(start);
    disconnect_all(state, nstate);
    result->xacts = 0;
    result->latencies = 0;
    result->sqlats = 0;
    for (i = 0; i < nstate; i++) {
        result->xacts += state[i].cnt;
        result->latencies += state[i].txn_latencies;
        result->sqlats += state[i].txn_sqlats;
    }
    result->throttle_lag = thread->throttle_lag;
    result->throttle_lag_max = thread->throttle_lag_max;
    INSTR_TIME_SET_CURRENT(end);
    INSTR_TIME_ACCUM_DIFF(result->conn_time, end, start);
    if (logfile)
        fclose(logfile);
    return result;
}

/*
 * Support for duration option: set timer_exceeded after so many seconds.
 */

#ifndef WIN32

static void handle_sig_alarm(SIGNAL_ARGS)
{
    timer_exceeded = true;
}

static void setalarm(int seconds)
{
    pqsignal(SIGALRM, handle_sig_alarm);
    alarm(seconds);
}

#ifndef ENABLE_THREAD_SAFETY

/*
 * implements pthread using fork.
 */

typedef struct fork_pthread {
    pid_t pid;
    int pipes[2];
} fork_pthread;

static int pthread_create(pthread_t* thread, pthread_attr_t* attr, void* (*start_routine)(void*), void* arg)
{
    fork_pthread* th = NULL;
    void* ret = NULL;

    th = (fork_pthread*)xmalloc(sizeof(fork_pthread));
    if (pipe(th->pipes) < 0) {
        free(th);
        return errno;
    }

    th->pid = fork();
    if (th->pid == -1) /* error */
    {
        free(th);
        return errno;
    }
    if (th->pid != 0) /* in parent process */
    {
        close(th->pipes[1]);
        *thread = th;
        return 0;
    }

    /* in child process */
    close(th->pipes[0]);

    /* set alarm again because the child does not inherit timers */
    if (duration > 0)
        setalarm(duration);

    ret = start_routine(arg);
    write(th->pipes[1], ret, sizeof(TResult));
    close(th->pipes[1]);
    free(th);
    exit(0);
}

static int pthread_join(pthread_t th, void** thread_return)
{
    int status;

    while (waitpid(th->pid, &status, 0) != th->pid) {
        if (errno != EINTR)
            return errno;
    }

    if (thread_return != NULL) {
        /* assume result is TResult */
        *thread_return = xmalloc(sizeof(TResult));
        if (read(th->pipes[0], *thread_return, sizeof(TResult)) != sizeof(TResult)) {
            free(*thread_return);
            *thread_return = NULL;
        }
    }
    close(th->pipes[0]);

    free(th);
    return 0;
}
#endif
#else /* WIN32 */

static VOID CALLBACK win32_timer_callback(PVOID lpParameter, BOOLEAN TimerOrWaitFired)
{
    timer_exceeded = true;
}

static void setalarm(int seconds)
{
    HANDLE queue;
    HANDLE timer;

    /* This function will be called at most once, so we can cheat a bit. */
    queue = CreateTimerQueue();
    if (seconds > ((DWORD)-1) / 1000 || !CreateTimerQueueTimer(&timer,
                                            queue,
                                            win32_timer_callback,
                                            NULL,
                                            seconds * 1000,
                                            0,
                                            WT_EXECUTEINTIMERTHREAD | WT_EXECUTEONLYONCE)) {
        fprintf(stderr, "Failed to set timer\n");
        exit(1);
    }
}

/* partial pthread implementation for Windows */

typedef struct win32_pthread {
    HANDLE handle;
    void* (*routine)(void*);
    void* arg;
    void* result;
} win32_pthread;

static unsigned __stdcall win32_pthread_run(void* arg)
{
    win32_pthread* th = (win32_pthread*)arg;

    th->result = th->routine(th->arg);

    return 0;
}

static int pthread_create(pthread_t* thread, pthread_attr_t* attr, void* (*start_routine)(void*), void* arg)
{
    int save_errno;
    win32_pthread* th = NULL;

    th = (win32_pthread*)xmalloc(sizeof(win32_pthread));
    th->routine = start_routine;
    th->arg = arg;
    th->result = NULL;

    th->handle = (HANDLE)_beginthreadex(NULL, 0, win32_pthread_run, th, 0, NULL);
    if (th->handle == NULL) {
        save_errno = errno;
        free(th);
        return save_errno;
    }

    *thread = th;
    return 0;
}

static int pthread_join(pthread_t th, void** thread_return)
{
    if (th == NULL || th->handle == NULL)
        return errno = EINVAL;

    if (WaitForSingleObject(th->handle, INFINITE) != WAIT_OBJECT_0) {
        _dosmaperr(GetLastError());
        return errno;
    }

    if (thread_return)
        *thread_return = th->result;

    CloseHandle(th->handle);
    free(th);
    return 0;
}

#endif /* WIN32 */
